package org.ramoths.netcom.sessions;

import org.ramoths.netcom.sessions.SessionCommunication;
import org.ramoths.netcom.sessions.exceptions.IOExceptionDistante;
import org.ramoths.netcom.sessions.exceptions.IOExceptionFermeture;
import org.ramoths.netcom.sessions.verrous.VrDonneesDistantes;
import org.ramoths.netcom.sessions.verrous.VrDonneesInputStream;

import java.io.IOException;
import java.io.InputStream;

public class InputStreamConsommateur extends InputStream
{
	private final SessionCommunication sessionComm;
	private final int idInputStream;
	//
	private final byte[] donnees;
	private int indexe;
	private int taille;
	private boolean fin;
	private IOException exception;
	private final VrDonneesInputStream vrDonneesInputStream;
	//
	private byte[] donneesArrivees;
	private boolean finArrivee;
	private Exception exceptionArrive;
	private final VrDonneesDistantes vrDonneesDistantes;
	//
	private final float tauxMinimum;
	private final ThInputStreamConsommateur thInputStreamConsommateur;
	
	public InputStreamConsommateur(SessionCommunication sessionComm, int idInputStream, int tailleTempon, float tauxMinimum)
	{
		this.sessionComm = sessionComm;
		this.idInputStream = idInputStream;
		//
		this.donnees = new byte[tailleTempon];
		this.indexe = 0;
		this.taille = 0;
		this.fin = false;
		this.exception = null;
		this.vrDonneesInputStream = new VrDonneesInputStream();
		//
		this.donneesArrivees = null;
		this.finArrivee = false;
		this.exceptionArrive = null;
		this.vrDonneesDistantes = new VrDonneesDistantes();
		//
		this.tauxMinimum = tauxMinimum;
		//
		this.thInputStreamConsommateur = new ThInputStreamConsommateur(this);
		this.thInputStreamConsommateur.start();
	}
	
	@Override
	public int read() throws IOException
	{
		vrDonneesInputStream.lock();
		try
		{
			while (taille == 0 && !fin && exception == null)
			{
				vrDonneesInputStream.getConditionTauxMinimumAtteint().signal();
				vrDonneesInputStream.getConditionDonneesDisponibles().awaitUninterruptibly();
			}
			if (!fin && exception == null && (float) taille / donnees.length <= tauxMinimum)
				vrDonneesInputStream.getConditionTauxMinimumAtteint().signal();
			if (taille != 0)
			{
				int r = (donnees[indexe] >= 0 ) ? donnees[indexe] : 256 + donnees[indexe];
				indexe = (indexe + 1) % donnees.length;
				taille -= 1;
				return r;
			}
			else if (exception != null)
			{
				throw exception;
			}
			else // Fin
			{
				return -1;
			}
		}
		finally
		{
			vrDonneesInputStream.unlock();
		}
	}
	
	void runThInputStreamConsommateur()
	{
		int tailleADemander = 0;
		while (true) 
		{
			/*
			 * Attente de déclanchement
			 */
			if (Thread.interrupted())
				break;
			vrDonneesInputStream.lock();
			try
			{
				try
				{
					if ((float) taille / donnees.length > tauxMinimum)
						vrDonneesInputStream.getConditionTauxMinimumAtteint().await();
				}
				catch (InterruptedException ex)
				{
					break;
				}
				//
				tailleADemander = donnees.length - taille;
				//
				if (tailleADemander == 0)
					continue;
			}
			finally
			{
				vrDonneesInputStream.unlock();
			}
			/*
			 * Envoi de demande et attente de réponse
			 */
			vrDonneesDistantes.lock();
			try
			{
				sessionComm.iscDemanderPortion(idInputStream, tailleADemander);
				vrDonneesDistantes.getConditionDonneesVenues().awaitUninterruptibly();
			}
			finally
			{
				vrDonneesDistantes.unlock();
			}
			/*
			 * Intégration de la réponse
			 */
			vrDonneesInputStream.lock();
			try
			{
				if (donneesArrivees != null)
				{
					if (((indexe + taille) % donnees.length) + donneesArrivees.length <= donnees.length)
					{
						System.arraycopy(
								donneesArrivees,
								0,
								donnees,
								(indexe + taille) % donnees.length,
								donneesArrivees.length
								);
					}
					else
					{
						System.arraycopy(
								donneesArrivees,
								0,
								donnees,
								(indexe + taille) % donnees.length,
								donnees.length - ((indexe + taille) % donnees.length)
								);
						System.arraycopy(
								donneesArrivees,
								donnees.length - ((indexe + taille) % donnees.length),
								donnees,
								0,
								donneesArrivees.length - (donnees.length - ((indexe + taille) % donnees.length))
								);
					}
					taille += donneesArrivees.length;
				}
				if (!fin)
					fin = finArrivee;
				if (exception == null && exceptionArrive != null)
					exception = new IOExceptionDistante("Exception distante!", exceptionArrive);
				//
				vrDonneesInputStream.getConditionDonneesDisponibles().signal();
				//
				donneesArrivees = null;
				finArrivee = false;
				exceptionArrive = null;
				//
				if (fin || exception != null)
					break;
			}
			finally
			{
				vrDonneesInputStream.unlock();
			}
		}
	}
	
	void setDonnees(byte[] donnees, boolean fin, Exception exception)
	{
		vrDonneesDistantes.lock();
		try
		{
			donneesArrivees = donnees;
			finArrivee = fin;
			exceptionArrive = exception;
			vrDonneesDistantes.getConditionDonneesVenues().signal();
		}
		finally
		{
			vrDonneesDistantes.unlock();
		}
	}
	
	@Override
	public int available() throws IOException
	{
		vrDonneesInputStream.lock();
		try
		{
			return taille;
		}
		finally
		{
			vrDonneesInputStream.unlock();
		}
	}
	
	@Override
	public void close() throws IOException
	{
		/*
		 * Interruption du thInputStreamConsommateur
		 */
		thInputStreamConsommateur.interrupt();
		boolean interruption = false;
		while (true)
		{
			try
			{
				thInputStreamConsommateur.join();
				break;
			}
			catch (InterruptedException ex)
			{
				interruption = true;
			}
		}
		if (interruption)
			Thread.currentThread().interrupt();
		/*
		 * Vérification
		 */
		vrDonneesInputStream.lock();
		try
		{
			if (fin || exception != null)
				return;
		}
		finally
		{
			vrDonneesInputStream.unlock();
		}
		/*
		 * Envoi d'ordre de fermeture de flux et attent de la réponse
		 */
		vrDonneesDistantes.lock();
		try
		{
			sessionComm.iscEnvoyerFermeture(idInputStream);
			vrDonneesDistantes.getConditionDonneesVenues().awaitUninterruptibly();
		}
		finally
		{
			vrDonneesDistantes.unlock();
		}
		/*
		 * Integration des donnees
		 */
		vrDonneesInputStream.lock();
		try
		{
			if (!fin)
				fin = finArrivee;
			//
			if (exceptionArrive != null)
				exception = new IOExceptionDistante("Exception distante!", exceptionArrive);
			else
				exception = new IOExceptionFermeture("Flux fermé!");
			//
			vrDonneesInputStream.getConditionDonneesDisponibles().signal();
		}
		finally
		{
			vrDonneesInputStream.unlock();
		}
	}
	
	int getIdInputStream()
	{
		return idInputStream;
	}
}
